Mocking _init_redis as a no-op means self._redis_subscriber is never replaced during a reconnect, so these tests can't exercise — and would stay green under — a lifecycle bug in the loop. In event_bridge.py, when subscribe() (line 169) raises, the except calls _init_redis() (line 221), which reassigns self._redis_subscriber = self._redis_client.pubsub() (line 69) to a fresh object without closing the failed one. The finally (lines 225–230) then closes self._redis_subscriber — now the fresh replacement — so the failed subscriber leaks and the next iteration attempts to subscribe using the freshly created but already closed subscriber. Whether Redis internally manages to reconnect from that state can depend on the Redis client implementation, but the lifecycle ordering is unquestionably wrong. Could you fix the ordering in production (close the failed subscriber before re-init, or don't close the fresh one) in its own commit, and make the reconnect side effect here install successive distinct subscriber mocks so the test asserts each retry subscribes on the newly created instance? As written the tests only check call counts, which pass either way. (Same applies to test_gives_up_after_max_retries below.)

Fixed in 9213f1f: the except handler now calls a new _close_subscriber() helper to close the failed subscriber (and drop the reference) before _init_redis() replaces it, and the per-iteration finally is gone — the loop closes the current subscriber exactly once on exit, so the fresh replacement is never closed prematurely. Both reconnect tests (test_subscribe_error_waits_and_reinitializes_redis, test_gives_up_after_max_retries) now install successive distinct subscriber mocks via the _init_redis side effect and assert that each retry subscribes on the newly created instance and that every failed subscriber is closed exactly once.

Asserting broadcast_event_from_notification is awaited once with the right args is a fine check of _process_single_event's local contract, but the AsyncMock removes the one boundary that's actually broken in production. _process_single_event (event_bridge.py:237) runs in a worker thread and drives the coroutine on a freshly created event loop (lines 247–255), while broadcast_event_from_notification awaits event_queue.put() (websocket_manager.py:138) on an asyncio.Queue (:89) whose broadcaster waiter lives on the API loop (:211). Waking that waiter from the worker thread isn't thread-safe — it raises under asyncio debug mode and, in production, mutates the API loop's ready-queue without waking it, so delivery is unreliable and the error is swallowed at websocket_manager.py:201. Please marshal onto the API loop (e.g. asyncio.run_coroutine_threadsafe) in a separate commit, and add one integration-style test with a real manager + active broadcaster + worker thread. The assertion here stays valid under that fix, so it needn't change — the gap is the missing coverage, not this test.

Fixed in 64006d6: WebSocketManager.connect() now records the running loop when it starts the broadcaster task (exposed as WebSocketManager.loop), and _process_single_event() submits the coroutine to that loop with asyncio.run_coroutine_threadsafe() and waits on the future. The private-loop path remains only as a fallback while no broadcaster loop exists yet — the queue has no waiters then, so driving it from the worker thread is safe. Added the requested integration-style test (test_worker_thread_event_reaches_broadcaster_loop): real WebSocketManager, active broadcaster on the test loop, _process_single_event() called from a worker thread via asyncio.to_thread, asserting delivery to a connected client. The existing unit tests keep their assertions unchanged and only set manager.loop = None so they keep exercising the fallback path.

This asserts subscribe and close are each called twice, i.e. it encodes "a get_message error resubscribes immediately" as the expected recovery — but that recovery is itself broken. On this path the inner loop breaks (event_bridge.py:203–205) and falls straight into the finally (225–230), which closes self._redis_subscriber without calling _init_redis(), so the next outer iteration subscribes on the same, now-closed subscriber (the same lifecycle-ordering issue as the subscribe-failure path, minus the re-init). Layered on top of that is a separate defect: this path never increments retry_count and never waits — the max_retries/retry_delay back-off only guards the outer subscribe exception path (207–217), and a successful resubscribe resets retry_count to 0 (line 171), so a persistently-failing get_message busy-loops forever with no delay. If you fix these in separate commits (recreate the subscriber before resubscribing, and route this path through the same bounded back-off), this test's exact call counts will need to move with the fix — worth updating the assertion at the same time so it documents the hardened behavior rather than the current one.

Fixed in ab74671 (on top of 9213f1f): the get_message error is now re-raised into the outer handler instead of breaking, so the path runs through the same bounded back-off as a failing subscribe() — the failed subscriber is closed, retry_count is incremented and capped by max_retries, the loop waits retry_delay, and the resubscribe happens on the fresh subscriber from _init_redis(). One deviation from the suggested split: after 9213f1f fixes the except-handler lifecycle, the single re-raise delivers both the subscriber recreation and the bounded back-off, so both defects land in one commit rather than two. The test is renamed to test_get_message_error_reconnects_with_backoff and asserts the back-off wait, the retry accounting and that the resubscribe happens on the newly installed instance.

_event_queue is an unbounded queue.Queue() (event_bridge.py:30), so put_nowait never raises queue.Full and the except queue.Full handler (133–134) is unreachable in production — this test only reaches it by patching put_nowait to raise. Fine to keep as a guard, but could you note in the test name/docstring that it covers a defensive branch unreachable with the current unbounded queue, so it isn't read as real drop behavior?

Done — test_full_queue_drops_event_with_warning now carries a docstring stating it covers a defensive branch that is unreachable with the current unbounded queue.Queue and is only reachable by patching put_nowait, so it isn't read as real drop behavior. Folded into the test commit (79305cb).